package com.webchat.ugc.service.mall.im;


import com.webchat.common.enums.RedisKeyEnum;
import com.webchat.common.enums.messagequeue.MessageQueueEnum;
import com.webchat.common.service.RedisService;
import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
import com.webchat.common.util.JsonUtil;
import com.webchat.domain.vo.dto.mall.im.UgcServerMessageDTO;
import com.webchat.ugc.mongodb.document.MallChatMessage;
import com.webchat.ugc.mongodb.document.SenderMessageCount;
import com.webchat.ugc.mongodb.inter.IMallChatMessageRepository;
import com.webchat.ugc.service.mall.MallShopService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationResults;
import org.springframework.data.mongodb.core.aggregation.GroupOperation;
import org.springframework.data.mongodb.core.aggregation.MatchOperation;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;


/**
 * 商家智能客服对话列表服务
 *
 *
 */
@Service
public class MallChatMessageService {


    @Autowired
    private RedisService redisService;

    @Autowired
    private IMallChatMessageRepository messageRepository;

    @Autowired
    private MessageQueueProducer<UgcServerMessageDTO, Integer> messageQueueProducer;

    @Autowired
    private MallShopService mallShopService;

    @Autowired
    private MongoTemplate mongoTemplate;

    /**
     * 最新60条数据作为热点数据，初始化到redis
     *
     */
    private static final int HOT_MESSAGE_COUNT = 60;


    /**
     * 查询两个账号之间历史对话记录（对话时间倒序查询）
     *
     * @param loginAccount
     * @param chatAccount
     * @param lastTime
     * @param size
     * @return
     */
    public List<UgcServerMessageDTO> list(String loginAccount, String chatAccount, Long lastTime, int size, int loadScreenNo) {
        // 判断是否买家，user 买家，shop 商家
        Long shopId = mallShopService.getManageShopId(loginAccount);
        String relationAccount = loginAccount;
        boolean isUserAccount = shopId == null;
        if (!isUserAccount) {
            // 关联最新创建店铺(默认 1:1 开店)
            relationAccount = String.valueOf(shopId);
        }
        return this.list(loginAccount, relationAccount, isUserAccount, chatAccount, lastTime, size, loadScreenNo);
    }

    public List<UgcServerMessageDTO> list(String loginAccount, String relationAccount, boolean isUserAccount,
                                          String chatAccount, Long lastTime, int size, int loadScreenNo) {

        lastTime = lastTime == null ? System.currentTimeMillis() : lastTime;
        String cacheKey = messageHistoryRedisCacheKey(relationAccount, chatAccount);
        // 判断热点缓存是否失效，失效则初始化
        doInitWithRedisNoneCache(cacheKey, relationAccount, chatAccount);
        // 计算本次查询热点数据还是冷数据
        Long cacheSize = redisService.zsize(cacheKey);
        int redisCacheSize = cacheSize == null ? 0 : cacheSize.intValue();
        boolean loadByRedis = loadScreenNo * size <= redisCacheSize;
        // 查询IM对话消息数据
        if (loadByRedis) {
            // 先走redis查询近期对话数据
            return queryMessageListFromRedis(cacheKey, lastTime, size);
        }
        // 如果超出热点数据缓存范围，走mongo读冷数据
        return queryMessageListFromMongo(relationAccount, chatAccount, lastTime, size);
    }

    public List<UgcServerMessageDTO> listOrderBySendTimeAsc(String relationAccount,
                                                            String chatAccount, Long msgTime, Long endTime) {

        String cacheKey = messageHistoryRedisCacheKey(relationAccount, chatAccount);
        // 判断热点缓存是否失效，失效则初始化
        doInitWithRedisNoneCache(cacheKey, relationAccount, chatAccount);
        // 计算本次查询热点数据还是冷数据
        boolean loadByRedis = redisService.zExistScore(cacheKey, msgTime);
        // 查询IM对话消息数据
        if (loadByRedis) {
            // 先走redis查询近期对话数据
            return queryMessageListFromRedisOrderByScoreAsc(cacheKey, msgTime, endTime);
        }
        // 如果超出热点数据缓存范围，走mongo读冷数据
        // 查询账号之间指定之间之后的1000条消息
        int maxSize = Math.min(Integer.MAX_VALUE, 1000);
        return queryMessageListFromMongoOrderBySendTimeAsc(relationAccount, chatAccount, msgTime, maxSize);
    }

    /**
     * 走redis 瀑布流式加载历史对话消息
     *
     * @param cacheKey
     * @param lastTime
     * @param size
     * @return
     */
    private List<UgcServerMessageDTO> queryMessageListFromRedis(String cacheKey, Long lastTime, int size) {

        Set<String> messageCache = redisService.zreverseRangeByScore(cacheKey, 0, lastTime, 0, size);
        if (CollectionUtils.isEmpty(messageCache)) {
            return Collections.emptyList();
        }
        return messageCache.stream().map(msg ->
                JsonUtil.fromJson(msg, UgcServerMessageDTO.class))
                .sorted(Comparator.comparing(UgcServerMessageDTO::getMsgTime)) // 升序
                .toList();
    }


    private List<UgcServerMessageDTO> queryMessageListFromRedisOrderByScoreAsc(String cacheKey, long startScore, long endScore) {

        Set<String> messageCache = redisService.zrangeByScore(cacheKey, startScore, endScore);
        if (CollectionUtils.isEmpty(messageCache)) {
            return Collections.emptyList();
        }
        return messageCache.stream().map(msg ->
                JsonUtil.fromJson(msg, UgcServerMessageDTO.class)).toList();
    }

    /**
     * 走redis 瀑布流式翻页加载历史对话消息
     *
     * @param account
     * @param chatAccount
     * @param lastTime
     * @param size
     * @return
     */
    private List<UgcServerMessageDTO> queryMessageListFromMongo(String account, String chatAccount,
                                                                Long lastTime, int size) {

        return queryMessageListFromMongo(account, chatAccount, lastTime, size, Sort.Direction.ASC, "$lte");
    }

    private List<UgcServerMessageDTO> queryMessageListFromMongoOrderBySendTimeAsc(String account, String chatAccount,
                                                                Long lastTime, int size) {

        return queryMessageListFromMongo(account, chatAccount, lastTime, size, Sort.Direction.ASC, "$gte");
    }

    private List<UgcServerMessageDTO> queryMessageListFromMongo(String account, String chatAccount,
                                                                Long lastTime, int size,
                                                                Sort.Direction direction,
                                                                String sendDateCondition) {
        Date lastSendDate = lastTime == null ? new Date() : new Date(lastTime);
        Pageable pageable = PageRequest.of(0, size, Sort.by(direction, "sendDate"));
        List<MallChatMessage> messages = messageRepository.findChatMessagesBetweenUsersBeforeDate(
                account,
                chatAccount,
                lastSendDate,
                sendDateCondition,
                pageable);
        if (CollectionUtils.isEmpty(messages)) {
            return Collections.emptyList();
        }
        return messages.stream().map(msg -> {
            UgcServerMessageDTO msgDTO = new UgcServerMessageDTO();
            BeanUtils.copyProperties(msg, msgDTO);
            msgDTO.setMsgTime(msg.getSendDate().getTime());
            msgDTO.setMessage(msg.getContent());
            return msgDTO;
        }).toList();
    }
    /**
     * 批量查询跟当前账号之间对话一组账号之间的按照消息发送时间倒序的{size}条消息数据
     *
     * @param account
     * @param chatAccounts
     * @param lastTime
     * @param size
     * @return
     */
    private Map<String, List<UgcServerMessageDTO>> queryMessageListFromMongo(String account, List<String> chatAccounts,
                                                                             Long lastTime, int size) {
        if (CollectionUtils.isEmpty(chatAccounts)) {
            return Collections.emptyMap();
        }

        Map<String, List<UgcServerMessageDTO>> res = new HashMap<>();
        for (String chatAccount : chatAccounts) {
            List<UgcServerMessageDTO> messages = queryMessageListFromMongo(account, chatAccount, lastTime, size);
            res.put(chatAccount, messages);
        }
        return res;
    }

    /**
     * 预热缓存
     *
     * @param cacheKey
     * @param relationAccount
     * @param chatAccount
     */
    private void doInitWithRedisNoneCache(String cacheKey, String relationAccount, String chatAccount) {

        if (redisService.exists(cacheKey)) {
            return;
        }
        /**
         * 缓存失效后预热
         *
         */
        List<UgcServerMessageDTO> list = queryMessageListFromMongo(relationAccount, chatAccount, null, HOT_MESSAGE_COUNT);
        if (CollectionUtils.isEmpty(list)) {
            // TODO 需要考虑，缓存击穿问题
            return;
        }
        Set<ZSetOperations.TypedTuple<String>> typedTupleSet = new HashSet<>();
        for (UgcServerMessageDTO message : list) {
            String value = JsonUtil.toJsonString(message);
            double score = message.getMsgTime();
            ZSetOperations.TypedTuple<String> tuple = new DefaultTypedTuple<>(value, score);
            typedTupleSet.add(tuple);
        }
        redisService.zadd(cacheKey, typedTupleSet, RedisKeyEnum.MALL_IM_MESSAGE_HISTORY_CACHE.getExpireTime());
    }

    /**
     * 持久化用户消息
     *
     * @param ugcServerMessage
     */
    public void add(UgcServerMessageDTO ugcServerMessage) {

        // MSG：redis 缓存，缓存最新60条数据热点数据（3屏数据，超出数据后续走mongo读）
        addMessageHistoryRedisCache(ugcServerMessage);

        // MSG：monogodb 存储
        messageQueueProducer.send(MessageQueueEnum.UGC_PERSISTENT_MALL_IM_MESSAGE, ugcServerMessage);

        // UNREAD MSG：未读消息缓存 （HASH）
        setChatSessionReadStatus(ugcServerMessage.getSenderId(), ugcServerMessage.getReceiverId(), ugcServerMessage.getRead());
    }

    /**
     * 查询两个账号之间是否有新消息
     *
     * @param account      当前登录账号
     * @param chatAccount  当前账号的对话账号
     * @return
     */
    public boolean hasNewMessage(String account, String chatAccount) {

        String sessionReadStatusCacheKey = chatSessionReadStatusCacheKey(account);

        // 校验新消息缓存是否失效或者首次请求
        boolean isExist = redisService.hexists(sessionReadStatusCacheKey, chatAccount);
        if (!isExist) {
            // 初始化一次
            return initAccountChatSessionReadStatusCache(account, chatAccount);
        }
        String readStatus = redisService.hget(sessionReadStatusCacheKey, chatAccount);
        return !Boolean.parseBoolean(readStatus);
    }

    /**
     * 批量查询账号间的已读未读状态
     *
     * @param account
     * @param chatAccounts
     * @return
     */
    public Map<String, Boolean> batchGetNewMessageStatus(String account, List<String> chatAccounts) {

        Map<String, Boolean> statusMap = new HashMap<>();

        String sessionReadStatusCacheKey = chatSessionReadStatusCacheKey(account);

        // 批量查询hash redis 缓存
        List<String> caches = redisService.hmget(sessionReadStatusCacheKey, chatAccounts);

        List<String> noneCacheAccounts = new ArrayList<>();
        for (int i = 0; i < chatAccounts.size(); i++) {
            String chatAccount = chatAccounts.get(i);
            String cache = caches.get(i);
            if (StringUtils.isNotBlank(cache)) {
                statusMap.put(chatAccount, !Boolean.parseBoolean(cache));
            } else {
                noneCacheAccounts.add(chatAccount);
            }
        }

        if (CollectionUtils.isNotEmpty(chatAccounts)) {
            // 存在需要重新走mongo初始化状态的账号
            Map<String, Boolean> refreshCacheMap = batchInitChatSessionNewMessageStatus(account, chatAccounts);
            statusMap.putAll(refreshCacheMap);
        }
        return statusMap;
    }

    /**
     * 批量刷新账号未读新消息缓存
     *
     * @param account
     * @param chatAccounts
     * @return Map<String, Boolean> <对话账号，是否有新消息（true代表有未读的）>
     */
    private Map<String, Boolean> batchInitChatSessionNewMessageStatus(String account, List<String> chatAccounts) {

        // 1. 构建匹配条件
        Criteria criteria = Criteria.where("senderId").in(chatAccounts)
                            .and("receiverId").is(account)
                            .and("read").is(false);
        MatchOperation matchStage = Aggregation.match(criteria);
        // 2. 构建分组操作
        GroupOperation groupStage = Aggregation.group("senderId")
                .count().as("count");
        // 3. 执行聚合查询
        Aggregation aggregation = Aggregation.newAggregation(matchStage, groupStage);
        AggregationResults<SenderMessageCount> results = mongoTemplate.aggregate(
                aggregation,
                "mall_chat_message",
                SenderMessageCount.class
        );
        List<SenderMessageCount> senderMessageCounts = results.getMappedResults();
        Map<String, Boolean> resMap = new HashMap<>();
        Map<String, String> refreshCacheMap = new HashMap<>();
        senderMessageCounts.forEach(unread -> {
            boolean hasNewMsg = unread.getCount() > 0;
            resMap.put(unread.getSenderId(), hasNewMsg);
            refreshCacheMap.put(unread.getSenderId(), String.valueOf(!hasNewMsg));
        });
        // 批量刷新缓存
        batchInitAccountChatSessionReadStatusCache(account, refreshCacheMap);
        return resMap;
    }

    /**
     * 走mongo初始化redis未读状态缓存
     *
     * @param account
     * @param chatAccount
     *
     * @return 返回是否有新消息
     */
    private boolean initAccountChatSessionReadStatusCache(String account, String chatAccount) {

        /**
         * 查询account是否有来自chatAccount的未读消息（查chatAccount发给account的）
         */
        Long unReadMsgCount = messageRepository.countBySenderIdAndReceiverIdAndRead(chatAccount, account, Boolean.FALSE);
        String chatSessionReadStatusACacheKey = chatSessionReadStatusCacheKey(account);
        boolean hasNewMessage = unReadMsgCount > 0;
        redisService.hset(chatSessionReadStatusACacheKey, chatAccount, String.valueOf(hasNewMessage));
        return hasNewMessage;
    }

    /**
     * 批量更新用户未读状态缓存
     *
     * @param account
     * @param chatAccountReadMap <对话账号，已读未读状态（true代表全部已读，无新消息）>
     */
    private void batchInitAccountChatSessionReadStatusCache(String account, Map<String, String> chatAccountReadMap) {

        String readStatusCacheKey = chatSessionReadStatusCacheKey(account);
        redisService.hmSet(readStatusCacheKey, chatAccountReadMap, RedisKeyEnum.MALL_IM_MESSAGE_HISTORY_CACHE.getExpireTime());
    }

    /**
     * 更新会话新消息状态
     *
     * @param account
     * @param chatAccount
     * @param isRead
     */
    public void setChatSessionReadStatus(String account, String chatAccount, boolean isRead) {

        String sessionReadStatusCacheKey = chatSessionReadStatusCacheKey(account);
        if (redisService.exists(sessionReadStatusCacheKey)) {
            return;
        }
        redisService.hset(sessionReadStatusCacheKey, chatAccount, String.valueOf(isRead),
                          RedisKeyEnum.MALL_CHAT_SESSION_READ_STATUS_HASH.getExpireTime());
    }

    /**
     * 对话session新消息状态
     * @param account
     * @return
     */
    private String chatSessionReadStatusCacheKey(String account) {

        return RedisKeyEnum.MALL_CHAT_SESSION_READ_STATUS_HASH.getKey(account);
    }

    /**
     * 消息数据持久化到mongodb
     *
     * @param message MQ消费到的对话消息
     * @return
     */
    public boolean doSaveMessage2Mongo(String message) {

        if (StringUtils.isBlank(message)) {
            return true;
        }
        UgcServerMessageDTO ugcServerMessageDTO = JsonUtil.fromJson(message, UgcServerMessageDTO.class);
        messageRepository.save(covertToMallChatMessage(ugcServerMessageDTO));
        return true;
    }

    /**
     * 批量查询与当前登录账号对话的一批次之间最后一条对话消息内容
     *
     * @param account
     * @param chatAccounts
     * @return
     */
    public Map<String, UgcServerMessageDTO<Object>> batchGetChatLastMessage(String account, List<String> chatAccounts) {

        if (CollectionUtils.isNotEmpty(chatAccounts)) {
            chatAccounts = chatAccounts.stream().filter(chatAccount -> !account.equals(chatAccount)).toList();
        }
        if (CollectionUtils.isEmpty(chatAccounts)) {
            return Collections.emptyMap();
        }

        Map<String, UgcServerMessageDTO<Object>> lastChatMessageRes = new HashMap<>();

        List<String> keys = chatAccounts.stream().map(
                chatAccount -> messageHistoryRedisCacheKey(account, chatAccount)).toList();
        Map<String, Set<String>> lastChatMessageMap =
                redisService.zreverseRangeByPipelined(keys, chatAccounts, 0, 1);

        // 缓存失效的账号
        List<String> noneCacheAccounts = new ArrayList<>();

        if (MapUtils.isNotEmpty(lastChatMessageMap)) {
            for (Map.Entry<String, Set<String>> entry : lastChatMessageMap.entrySet()) {
                String chatAccount = entry.getKey();
                Set<String> messages = entry.getValue();
                UgcServerMessageDTO<Object> lastMessage;
                if (CollectionUtils.isNotEmpty(messages)) {
                    // 缓存中成功获取到消息数据
                    String lastMessageJson = new ArrayList<>(messages).get(0);
                    lastMessage = JsonUtil.fromJson(lastMessageJson, UgcServerMessageDTO.class);
                    lastChatMessageRes.put(chatAccount, lastMessage);
                } else {
                    // 缓存失效的账号集合，需要重新查询mongo
                    noneCacheAccounts.add(chatAccount);
                }
            }
        }

        // 针对redis失效的对话消息账号，批量查询mongo重新预热（保证接下来对话账号列表消息数据可以走redis获取）
        if (CollectionUtils.isNotEmpty(noneCacheAccounts)) {
            Map<String, List<UgcServerMessageDTO>> noneCacheAccountMessage =
                    this.queryMessageListFromMongo(account, noneCacheAccounts, null, HOT_MESSAGE_COUNT);
            // 重新讲热消息数据写到redis
            this.doInitChatHotMessageRedisCache(account, noneCacheAccountMessage);
            // 追加消息到返回结果集
            for (Map.Entry<String, List<UgcServerMessageDTO>> entry : noneCacheAccountMessage.entrySet()) {
                // TODO
                if (CollectionUtils.isNotEmpty(entry.getValue())) {
                    lastChatMessageRes.put(entry.getKey(), entry.getValue().get(0));
                }
            }
        }
        return lastChatMessageRes;
    }

    private void doInitChatHotMessageRedisCache(String account, Map<String, List<UgcServerMessageDTO>> noneCacheAccountMessage) {

        for (Map.Entry<String, List<UgcServerMessageDTO>> entry : noneCacheAccountMessage.entrySet()) {

            String cacheKey = messageHistoryRedisCacheKey(account, entry.getKey());
            List<UgcServerMessageDTO> chatMessages = entry.getValue();

            Set<ZSetOperations.TypedTuple<String>> tuples = chatMessages.stream().map(msg -> {
                String value = JsonUtil.toJsonString(msg);
                double score = msg.getMsgTime();
                ZSetOperations.TypedTuple<String> typedTuple = new DefaultTypedTuple(value, score);
                return typedTuple;
            }).collect(Collectors.toSet());
            redisService.zadd(cacheKey, tuples, RedisKeyEnum.MALL_CHAT_ACCOUNT_ZSET_CACHE.getExpireTime());
        }
    }

    /**
     * IM 对话消息缓存 （ZSET）
     *
     * @param ugcServerMessage
     */
    private void addMessageHistoryRedisCache(UgcServerMessageDTO ugcServerMessage) {

        String sender2ReceiverCacheKey = messageHistoryRedisCacheKey(ugcServerMessage.getSenderId(), ugcServerMessage.getReceiverId());
        String receiver2SenderCacheKey = messageHistoryRedisCacheKey(ugcServerMessage.getReceiverId(), ugcServerMessage.getSenderId());
        doAddMessageHistoryRedisCache(sender2ReceiverCacheKey, ugcServerMessage);
        doAddMessageHistoryRedisCache(receiver2SenderCacheKey, ugcServerMessage);
    }

    /**
     * 执行消息数据缓存
     *
     * @param cacheKey
     * @param ugcServerMessage
     */
    private void doAddMessageHistoryRedisCache(String cacheKey, UgcServerMessageDTO ugcServerMessage) {

        String messageJson = JsonUtil.toJsonString(ugcServerMessage);
        if (redisService.exists(cacheKey)) {
            // 缓存存在，或者未失效，直接追加
            redisService.zadd(cacheKey, messageJson, ugcServerMessage.getMsgTime());
            redisService.expire(cacheKey, RedisKeyEnum.MALL_CHAT_ACCOUNT_ZSET_CACHE.getExpireTime());
        }
    }

    private String messageHistoryRedisCacheKey(String account, String chatAccount) {

        return RedisKeyEnum.MALL_IM_MESSAGE_HISTORY_CACHE.getKey(account, chatAccount);
    }


    /**
     * 撤回消息
     *
     * @param ugcServerMessage
     */
    public void revoke(UgcServerMessageDTO ugcServerMessage) {

    }

    /**
     * 删除消息
     *
     * @param ugcServerMessage
     */
    public void delete(UgcServerMessageDTO ugcServerMessage) {

    }


    /**
     * IM 消息 DTO -> mongo entity
     *
     * @param ugcServerMessage
     * @return
     */
    private MallChatMessage covertToMallChatMessage(UgcServerMessageDTO ugcServerMessage) {
        MallChatMessage mallChatMessage = new MallChatMessage();
        mallChatMessage.setSenderId(ugcServerMessage.getSenderId());
        mallChatMessage.setReceiverId(ugcServerMessage.getReceiverId());
        mallChatMessage.setProxySenderId(ugcServerMessage.getProxySenderId());
        mallChatMessage.setProxyReceiverId(ugcServerMessage.getProxyReceiverId());
        mallChatMessage.setRead(ugcServerMessage.getRead());
        mallChatMessage.setSendDate(new Date(ugcServerMessage.getMsgTime()));
        mallChatMessage.setMessageType(ugcServerMessage.getMessageType());
        mallChatMessage.setContent(ugcServerMessage.getMessage());
        return mallChatMessage;
    }
}
